package event;
import java.io.IOException;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.NavigableMap;
import java.util.Random;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import parallel.BasicRangeInputSplit;
public class SequenceGenerator extends Generator{
	private Mapper<LongWritable,NullWritable,LongWritable,Text>.Context context = null;
	private long currTime;
	private Map<String,Map<String,List<String>>> datastore = new HashMap<String, Map<String,List<String>>>();
	private Map<String, Long> startMap = new HashMap<String, Long>();
	private Map<String, Long> rangeMap = new HashMap<String, Long>();
	private Sequence sequence = null;
	public SequenceGenerator(long currTime, Mapper<LongWritable,NullWritable,LongWritable,Text>.Context context, BasicRangeInputSplit inputSplit, int numOfRegions) throws IOException
	{
		super(numOfRegions, inputSplit);
		this.inputSplit = inputSplit;
		this.currTime = currTime; 
		this.context = context;
		this.numOfRegions = numOfRegions;
		sequence = eventConfig.getSequence();
		getRange(inputSplit.getSplitId());
		loadDatastore();
	}
	private void loadDatastore() throws IOException
	{
		// region mode
		if(mode == 1)
		{
			List<Basic> basicList = basicConfig.getBasicList();
			List<EventNode> nodeList = nodeConfig.getEventNodeList();
			for(int i = 0; i < basicList.size(); i++)
			{
				Basic basic = basicList.get(i);
				EventNode  eventNode = nodeList.get(i);
				loadTable(basic, eventNode);
			}
		}
		// single level mode
		else if (mode == 2)
		{
			String homeNode = sequence.getHome();
			Basic basic = basicConfig.getBasic(homeNode);
			EventNode eventNode = nodeConfig.getEventNode(homeNode);
			loadTable(basic, eventNode);
		}
	}
	// get the links from HBase to a local map structure
	private void loadTable(Basic basic, EventNode eventNode) throws IOException
	{
		String name = basic.getName();
		long start = startMap.get(name);
		long range = rangeMap.get(name);
		long end = start + range;
		DecimalFormat formatter = new java.text.DecimalFormat("000000000000");
		// calculate the start row key and end row key of this split
		String startRowKey = name + "-" + formatter.format(start);
		String endRowKey = name + "-" + formatter.format(end);
		byte[] startRow = Bytes.toBytes(startRowKey);
		byte[] endRow = Bytes.toBytes(endRowKey);
		Configuration conf = HBaseConfiguration.create();
		HTable table = new HTable(conf, "Node");
		Scan scan = new Scan();
		scan.setStartRow(startRow);
		scan.setStopRow(endRow);
		ResultScanner scanner = table.getScanner(scan);
		List<Link> linkList = eventNode.getLinkList();
		// load the links of one page
		for (Result result : scanner) {
			byte [] rowKey = result.getRow();
			String key = Bytes.toString(rowKey);
			Map<String, List<String>> familyColumns = new HashMap<String, List<String>>();
			for(int j = 0; j < linkList.size(); j++)
			{
				Link link = linkList.get(j);
				String linkName = link.getLinkName();
				byte [] family = Bytes.toBytes(linkName);
				NavigableMap<byte [], byte[]> columnValues = result.getFamilyMap(family);
				List<String> columnQualifiers = new ArrayList<String>();
				for(byte [] column : columnValues.keySet())
				{
					columnQualifiers.add(Bytes.toString(column));
				}
				familyColumns.put(linkName, columnQualifiers);
			}
			datastore.put(key, familyColumns);			
		}
		scanner.close();
		table.close();
	}
	public void generate(long sessionId, String ignored) throws IOException, InterruptedException
	{
		if(mode == 2)
		{
			generateSingleLevel(sessionId);
		}
		else if(mode == 0)
		{
			generateAll(sessionId);
		}
		else
		{
			generateByRegion(sessionId);
		}
	}
	// Generate the sequences in navigation mode
	private void generateAll(long sessionId) throws IOException, InterruptedException
	{
		Random random = new Random();
		int length = getSequenceLength(sequence.getMinLength(), sequence.getMaxLength());
		String prevNode = null;
		String currNode = null;
		// time the initializer of the click will stay on one page
		int timeStep = 0;
		String homeName = sequence.getHome();
		Basic basic = basicConfig.getBasic(homeName);
		int homeMaxTimeStep = basic.getMaxTimeStep();
		int splitId = inputSplit.getSplitId();
		long currId = getId(homeName,0,basic.getNumber());
		DecimalFormat formatter = new java.text.DecimalFormat("000000000000");
		String id = formatter.format(currId);
		String homeId = homeName + "-" + id;
		int distance  = 1;
		for(int i = 0; i < length; i++)
		{
			StringBuffer sb = new StringBuffer();
			// generate timetamp
			currTime += timeStep;
			sb.append(homeId + ",");
			sb.append(splitId + "_" + sessionId +",");
			if(prevNode == null)
			{
				currNode = homeId;
				timeStep = random.nextInt(homeMaxTimeStep);
			}
			else
			{
				// the farther the user is away from his homepage, the smaller the possibity
				// that he will move forward instead of going back to his homepage
				double moveForward = 1/distance;
				double prob = random.nextDouble();
				// if the user moves forward
				if(prob <= moveForward)
				{
					String currEvent = currNode.split("-")[0]; 
					double prevProb = 0;
					prob = random.nextDouble();
					EventNode eventNode = nodeConfig.getEventNode(currEvent);
					// probabilities of clicking on different type of links
					Map<Double,String> distribs = new TreeMap<Double,String>();
					List<Link> linkList = eventNode.getLinkList();
					double cumulativeRatio = 0;
					for(int j = 0; j < linkList.size(); j++)
					{
						Link link = linkList.get(j);
						cumulativeRatio += link.getRatio();
						distribs.put(cumulativeRatio, link.getLinkName());
					}
					String eventName = null;
					Iterator<Double> it = distribs.keySet().iterator();
					// randomly pick a link type
					while(it.hasNext())
					{
						double currProb = it.next();
						if(prob >= prevProb && prob <= currProb)
						{
							eventName = distribs.get(currProb);
							break;
						}
						prevProb = currProb;
					}
					// randomly pick a link from that link type
					Basic event = basicConfig.getBasic(currEvent);
					Configuration conf = HBaseConfiguration.create();
					HTable table = new HTable(conf, "Node");
					Get get = new Get(Bytes.toBytes(prevNode));
					byte [] family = Bytes.toBytes(eventName);
					get.addFamily(family);
					Result result = table.get(get);
					NavigableMap<byte [], byte[]> columnValues = result.getFamilyMap(family);
					List<String> columnQualifiers = new ArrayList<String>();
					System.out.println("currNode: " + currNode + "  eventName: " + eventName);
					for(byte [] column : columnValues.keySet())
					{
						columnQualifiers.add(Bytes.toString(column));
					}
					int size = columnQualifiers.size();
					int index = random.nextInt(size);
					currNode = columnQualifiers.get(index);
					table.close();
					timeStep = random.nextInt(event.getMaxTimeStep());
					//increase the distance the user away from his homepage 
					distance *= 2;
				}
				// go back to user's homepage
				else
				{
					currNode = homeId;
					timeStep = random.nextInt(homeMaxTimeStep);
					distance = 1;
				}
			}
			sb.append(currNode + ",");
			prevNode = currNode;
			context.write(new LongWritable(currTime), new Text(sb.toString()));
		}
		currTime += timeStep;
		StringBuffer sb = new StringBuffer();
		sb.append(homeId + ",");
		sb.append(splitId + "_" + sessionId +",");
		sb.append("exit" + ","); 
		context.write(new LongWritable(currTime), new Text(sb.toString()));
	}
	// region mode, very similar to navigation mode
	private void generateByRegion(long sessionId) throws IOException, InterruptedException
	{
		Random random = new Random();
		int length = getSequenceLength(sequence.getMinLength(), sequence.getMaxLength());
		String prevNode = null;
		String currNode = null;
		int timeStep = 0;
		String homeName = sequence.getHome();
		Basic basic = basicConfig.getBasic(homeName);
		int homeMaxTimeStep = basic.getMaxTimeStep();
		int splitId = inputSplit.getSplitId();
		long currId = getId(homeName);
		DecimalFormat formatter = new java.text.DecimalFormat("000000000000");
		String id = formatter.format(currId);
		String homeId = homeName + "-" + id;
		int distance  = 1;
		for(int i = 0; i < length; i++)
		{
			StringBuffer sb = new StringBuffer();
			currTime += timeStep;
			sb.append(homeId + ",");
			sb.append(splitId + "_" + sessionId +",");
			if(prevNode == null)
			{
				currNode = homeId;
				timeStep = random.nextInt(homeMaxTimeStep);
			}
			else
			{
				double moveForward = 1/distance;
				double prob = random.nextDouble();
				if(prob <= moveForward)
				{
					double prevProb = 0;
					prob = random.nextDouble();
					String currEvent = currNode.split("-")[0];
					EventNode eventNode = nodeConfig.getEventNode(currEvent);
					Map<Double,String> distribs = new TreeMap<Double,String>();
					List<Link> linkList = eventNode.getLinkList();
					double cumulativeRatio = 0;
					for(int j = 0; j < linkList.size(); j++)
					{
						Link link = linkList.get(j);
						cumulativeRatio += link.getRatio();
						distribs.put(cumulativeRatio, link.getLinkName());
					}
					String eventName = null;
					Iterator<Double> it = distribs.keySet().iterator();
					while(it.hasNext())
					{
						double currProb = it.next();
						if(prob >= prevProb && prob <= currProb)
						{
							eventName = distribs.get(currProb);
							break;
						}
						prevProb = currProb;
					}
					Basic event = basicConfig.getBasic(currEvent);
					Map<String, List<String>> result = datastore.get(prevNode);
					List<String> columnQualifiers = result.get(eventName);
					int size = columnQualifiers.size();
					int index = random.nextInt(size);
					currNode = columnQualifiers.get(index);
					timeStep = random.nextInt(event.getMaxTimeStep());
					distance *= 2;
				}
				else
				{
					currNode = homeId;
					timeStep = random.nextInt(homeMaxTimeStep);
					distance = 1;
				}
			}
			sb.append(currNode + ",");
			prevNode = currNode;
			context.write(new LongWritable(currTime), new Text(sb.toString()));
		}
		currTime += timeStep;
		StringBuffer sb = new StringBuffer();
		sb.append(homeId + ",");
		sb.append(splitId + "_" + sessionId +",");
		sb.append("exit" + ","); 
		context.write(new LongWritable(currTime), new Text(sb.toString()));

	}
	// single level mode, very similar to navigation mode
	private void generateSingleLevel(long sessionId) throws IOException, InterruptedException
	{
		Random random = new Random();
		Sequence sequence = eventConfig.getSequence();
		int length = getSequenceLength(sequence.getMinLength(), sequence.getMaxLength());
		String homeName = sequence.getHome();
		EventNode eventNode = nodeConfig.getEventNode(homeName);
		Map<Double,String> distribs = new TreeMap<Double,String>();
		List<Link> linkList = eventNode.getLinkList();
		double cumulativeRatio = 0;
		for(int i = 0; i < linkList.size(); i++)
		{
			Link link = linkList.get(i);
			cumulativeRatio += link.getRatio();
			distribs.put(cumulativeRatio, link.getLinkName());
		}
		int splitId = inputSplit.getSplitId();
		long currId = getId(homeName);
		DecimalFormat formatter = new java.text.DecimalFormat("000000000000");
		String id = formatter.format(currId);
		String homeId = homeName + "-" + id;
		String currNode = null;
		int timeStep = 0;
		for(int i = 0; i < length; i++)
		{
			StringBuffer sb = new StringBuffer();
			currNode = homeId;
			Basic event = null;
			if(i % 2 != 0)
			{
				double prevProb = 0;
				double prob = random.nextDouble();
				String eventName = null;
				Iterator<Double> it = distribs.keySet().iterator();
				while(it.hasNext())
				{
					double currProb = it.next();
					if(prob >= prevProb && prob <= currProb)
					{
						eventName = distribs.get(currProb);
						break;
					}
					prevProb = currProb;
				}
				Map<String, List<String>> result = datastore.get(currNode);
				List<String> columnQualifiers = result.get(eventName);
				int size = columnQualifiers.size();
				int index = random.nextInt(size);
				currNode = columnQualifiers.get(index);
				String currEvent = currNode.split("-")[0];
				event = basicConfig.getBasic(currEvent);
			}
			else
			{
				event = basicConfig.getBasic(homeName);
			}
			sb = new StringBuffer();
			sb.append(homeId + ",");
			sb.append(splitId + "_" + sessionId +",");
			sb.append(currNode + ",");
			timeStep = random.nextInt(event.getMaxTimeStep());
			currTime += timeStep;
			context.write(new LongWritable(currTime), new Text(sb.toString()));
		}
		currTime += timeStep;
		StringBuffer sb = new StringBuffer();
		sb.append(homeId + ",");
		sb.append(splitId + "_" + sessionId +",");
		sb.append("exit" + ","); 
		context.write(new LongWritable(currTime), new Text(sb.toString()));
	}
	// get the length of the sequences in normal distribution
	private int getSequenceLength(int min, int max)
	{
		Random random = new Random();
		double a = ((double)min + (double)max)/2;
		double b = ((double)max - (double)min)/6;
		int length = (int)Math.round(b * random.nextGaussian() + a);
		if(length < min)
			length = min;
		else if(length > max)
			length = max;
		return length;
	}
	// get the range of the split
	private void getRange(int splitId)
	{
		List<Basic> basicList = basicConfig.getBasicList();
		for(int i = 0; i < basicList.size(); i++)
		{
			Basic basic = basicList.get(i);
			String name = basic.getName();
			long total = basic.getNumber();
			long numPerRegion = total/numOfRegions;
			long start = numPerRegion * splitId, range;
			System.out.print("start:" + start + " total: " + total + " splitId: " + splitId);
			if(splitId < numOfRegions - 1)
			{
				range = numPerRegion;
			}
			else
			{
				range = total - start; 
			}
			System.out.print("range :" + range);
			startMap.put(name, start);
			rangeMap.put(name, range);
		}
	}
	// randomly pick a page id from a range
	private long getId(String name, long start, long range)
	{
		long id;
		Random random = new Random();
		if(start + range < Integer.MAX_VALUE)
		{
			id = (int)start + random.nextInt((int)(range));
		}
		else
		{
			id = start + Math.abs(random.nextLong()) % range;
		}
		return id;
	}
	private long getId(String name)
	{
		long start = startMap.get(name);
		long range = rangeMap.get(name);
		return getId(name, start, range);
	}
}
